package com.ydl.learning.flink.demo.transformations

import com.ydl.learning.flink.demo.Utils
import org.apache.flink.api.scala._

/**
 *
 *
 * @author ydl
 * @since 2020/10/12
 */
object JoinDemo extends App with Utils {
  val input1 = env.fromElements((1, 2), (2, 3), (3, 4))
  val input2 = env.fromElements((4, 1), (5, 2), (6, 4))
  //内连接 where和equalTo指定join的条件，where指定左表 equalTo指定右表
  input1.join(input2).where(0).equalTo(1).print()

  //外连接，需要指定值不存在时的表达式
  input1.leftOuterJoin(input2).where(0).equalTo(1) {
    (input1, input2) => (input1, if (input2 == null) (0, 0) else input2)
  }.print()
  input1.rightOuterJoin(input2).where(0).equalTo(1) {
    (input1, input2) => (if (input1 == null) (0, 0) else input1, input2)
  }.print()
  input1.fullOuterJoin(input2).where(0).equalTo(1) {
    (input1, input2) => (if (input1 == null) (0, 0) else input1, if (input2 == null) (0, 0) else input2)
  }.print()

}
